package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/3/14 09:17
 */
public class Dwd_03_DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeOrderDetail().init(
            3003,
            2,
            "Dwd_03_DwdTradeOrderDetail"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 从业务上来说, 4 张事实表之间没有时间的间隔, 5s 考虑的时候网络延迟
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
//        tEnv.getConfig().getConfiguration().setString("table.exec.state.ttl", "5 second");
        
        // 1. 先读取 ods_db
        readOdsDb(tEnv, "Dwd_03_DwdTradeOrderDetail");
        // 2. 读取字典表
        readBaseDic(tEnv);
        
        // 3. 过滤出 order_detail 数据: insert
        Table orderDetail = tEnv.sqlQuery("select " +
                                              "data['id'] id," +
                                              "data['order_id'] order_id," +
                                              "data['sku_id'] sku_id," +
                                              "data['sku_name'] sku_name," +
                                              "data['create_time'] create_time," +
                                              "data['source_id'] source_id," +
                                              "data['source_type'] source_type," +
                                              "data['sku_num'] sku_num," +
                                              "cast(cast(data['sku_num'] as decimal(16,2)) * " +
                                              "cast(data['order_price'] as decimal(16,2)) as String) split_original_amount," +
                                              "data['split_total_amount'] split_total_amount," +
                                              "data['split_activity_amount'] split_activity_amount," +
                                              "data['split_coupon_amount'] split_coupon_amount," +
                                              "ts," +
                                              "pt " +
                                              "from ods_db " +
                                              "where `database`='gmall2022' " +
                                              "and `table`='order_detail' " +
                                              "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
        
        //        orderDetail.execute().print();
        
        // 4. 过滤出 order_info 数据
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id, " +
                                            "data['user_id'] user_id, " +
                                            "data['province_id'] province_id " +
                                            "from ods_db " +
                                            "where `database`='gmall2022' " +
                                            "and `table`='order_info' " +
                                            "and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        
        
        // 5. 过滤出 order_detail_activity 数据
        Table orderDetailActivity = tEnv.sqlQuery("select " +
                                                      "data['order_detail_id'] order_detail_id, " +
                                                      "data['activity_id'] activity_id, " +
                                                      "data['activity_rule_id'] activity_rule_id " +
                                                      "from ods_db " +
                                                      "where `database`='gmall2022' " +
                                                      "and `table`='order_detail_activity' " +
                                                      "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail_activity", orderDetailActivity);
        
        
        // 6. 过滤出 order_detail_coupon 数据
        Table orderDetailCoupon = tEnv.sqlQuery("select " +
                                                    "data['order_detail_id'] order_detail_id, " +
                                                    "data['coupon_id'] coupon_id " +
                                                    "from ods_db " +
                                                    "where `database`='gmall2022' " +
                                                    "and `table`='order_detail_coupon' " +
                                                    "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);
        // 7. 5 张表进行 join
        Table result = tEnv.sqlQuery("select " +
                                        "od.id, " +
                                        "od.order_id, " +
                                        "oi.user_id, " +
                                        "od.sku_id, " +
                                        "od.sku_name, " +
                                        "oi.province_id, " +
                                        "act.activity_id, " +
                                        "act.activity_rule_id, " +
                                        "cou.coupon_id, " +
                                        "date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                                        "od.create_time, " +
                                        "od.source_id, " +
                                        "od.source_type, " +
                                        "dic.dic_name source_type_name, " +
                                        "od.sku_num, " +
                                        "od.split_original_amount, " +
                                        "od.split_activity_amount, " +
                                        "od.split_coupon_amount, " +
                                        "od.split_total_amount, " +
                                        "od.ts " +
                                        "from order_detail od " +
                                        "join order_info oi on od.order_id=oi.id " +
                                        "left join order_detail_activity act " +
                                        "on od.id=act.order_detail_id " +
                                        "left join order_detail_coupon cou " +
                                        "on od.id=cou.order_detail_id " +
                                        "join base_dic for system_time as of od.pt as dic " +
                                        "on od.source_type=dic.dic_code ");
        
        
        // 8. 把 join 的结果写出到 kafka(upsert-kafka) 中
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                          "id string, " +
                          "order_id string, " +
                          "user_id string, " +
                          "sku_id string, " +
                          "sku_name string, " +
                          "province_id string, " +
                          "activity_id string, " +
                          "activity_rule_id string, " +
                          "coupon_id string, " +
                          "date_id string, " +
                          "create_time string, " +
                          "source_id string, " +
                          "source_type string, " +
                          "source_type_name string, " +
                          "sku_num string, " +
                          "split_original_amount string, " +
                          "split_activity_amount string, " +
                          "split_coupon_amount string, " +
                          "split_total_amount string, " +
                          "ts bigint, " +
                          "primary key(id) not enforced " +
                          ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
    
        result.executeInsert("dwd_trade_order_detail");
    }
}
/*
交易域下单事务事实表
    粒度: 最小粒度  商品
    
    order_detail: insert
        join
    order_info: insert
        left join
    order_detail_activity: insert
        left join
    order_detail_coupon: insert
        lookup join
    base_dic
    
    

 */